//package com.study.iceberg.flink;
//
//import org.apache.flink.api.common.typeinfo.Types;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.table.data.RowData;
//import org.apache.iceberg.flink//.TableLoader;
//import org.apache.iceberg.flink.source.FlinkSource;
//
//public class JavaReadDemo1 {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);// 设置并行度，方便测试观察结果
//
//        /*
//        获取TableLoader，
//        fromHadoopTable获取hadoop catalog下的表
//        fromCatalog获取hive catalog下的表
//         */
//        TableLoader tableLoader = TableLoader
//                // hadoop catalog 的表可以直接通过表所在的hdfs目录获取
//                .fromHadoopTable("hdfs://192.168.200.101:9820/warehouse/iceberg-hadoop/iceberg_db/sample");
//
//        DataStream<RowData> inputDS = FlinkSource.forRowData()
//                .env(env)
//                .tableLoader(tableLoader)
//                //.streaming(false) // batch模式
//                .streaming(true) // streaming模式，不停的读取表数据
//                .build();
//
//        // 注意flink的rowdata的r.getString(1)返回的不是String类型,需要再toString一下
//        inputDS.map(r-> Tuple2.of(r.getInt(0),r.getString(1).toString()))
//                // java泛型擦除的问题
//                .returns(Types.TUPLE(Types.INT,Types.STRING))
//                .print();
//
//        env.execute();
//    }
//}
